use tokio::sync::broadcast;
use super::types::{IStreamReader, IStreamWriter, StreamData};
use async_trait::async_trait;

const BROADCAST_CHANNEL_CAPACITY: usize = 1024;

/// 流读取器的具体实现
pub struct StreamReader {
    receiver: broadcast::Receiver<StreamData>,
}

#[async_trait]
impl IStreamReader for StreamReader {
    /// 从广播频道接收数据
    async fn read(&mut self) -> Option<StreamData> {
        match self.receiver.recv().await {
            Ok(data) => Some(data),
            Err(broadcast::error::RecvError::Closed) => None, // 频道关闭
            Err(broadcast::error::RecvError::Lagged(_)) => {
                // 消费者速度过慢，丢失了一些消息。在某些场景下可能需要记录日志。
                // 我们选择继续尝试接收下一条消息。
                self.read().await
            }
        }
    }
}

/// 流写入器的具体实现
#[derive(Clone)]
pub struct StreamWriter {
    sender: broadcast::Sender<StreamData>,
}

#[async_trait]
impl IStreamWriter for StreamWriter {
    /// 向广播频道发送数据
    async fn write(&self, data: StreamData) -> Result<(), String> {
        self.sender.send(data).map(|_| ()).map_err(|e| e.to_string())
    }

    /// 关闭写入器（在 broadcast 模型中，写入器的关闭通常意味着整个流的结束，
    /// 但由于 sender 可以被克隆，我们在这里不直接关闭 channel，
    /// 而是让 MulticastStream 来管理生命周期）
    async fn close(&self) {
        // no-op
    }
}

/// 多播流的整体封装
pub struct MulticastStream {
    name: String,
    sender: broadcast::Sender<StreamData>,
}

impl MulticastStream {
    pub fn new(name: &str) -> Self {
        let (sender, _) = broadcast::channel(BROADCAST_CHANNEL_CAPACITY);
        Self {
            name: name.to_string(),
            sender,
        }
    }

    /// 添加一个新的读取器（订阅者）
    pub fn add_reader(&self) -> Box<dyn IStreamReader> {
        Box::new(StreamReader { receiver: self.sender.subscribe() })
    }

    /// 添加一个新的写入器（发布者）
    pub fn add_writer(&self) -> Box<dyn IStreamWriter> {
        Box::new(StreamWriter { sender: self.sender.clone() })
    }

    pub fn name(&self) -> &str {
        &self.name
    }
}